package com.zk.lock.lock1;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于zookeeper的EPHEMERAL_SEQUENTIAL类型节点及watch机制，实现分布式锁
 * 
 * 1 开始10个线程，在disLocks节点下各自创建名为sub的EPHEMERAL_SEQUENTIAL节点
 * 
 * 2 获取disLocks节点下的所有子节点，排序，如果自己的节点编号最小，则获取锁
 * 
 * 3 否则watch排在自己前面的节点，监听到其删除后，进入第2步(重新检测排序是防止监听的节点发生连接失效导致的连接删除情况)
 * 
 * 4 删除自身sub节点，释放连接
 * 
 * 
 *
 */
public class DistributedLock implements Watcher {

	private static Logger log = LoggerFactory.getLogger(DistributedLock.class);

	private int threadId;
	private ZooKeeper zk = null;
	private String selfPath;
	private String waitPath;
	private String LOG_PREFIX_OF_THREAD;

	private static final int SESSION_TIMEOUT = 10000;

	private static final String GROUP_PATH = "/disLocks";

	private static final String SUB_PATH = "/disLocks/sub";

	private static final String CONNECTION_STRING = "192.163.35.61:2181";

	private static final int THREAD_NUM = 10;

	/** 确保连接zk成功 */
	private CountDownLatch connectedSemaphore = new CountDownLatch(1);

	/** 确保所有线程运行结束 */
	private static CountDownLatch threadSemaphore = new CountDownLatch(THREAD_NUM);

	public DistributedLock(int id) {
		this.threadId = id;
		LOG_PREFIX_OF_THREAD = "第" + threadId + "个线程";
	}

	@Override
	public void process(WatchedEvent event) {
		if (event == null) {
			return;
		}
		Event.KeeperState keeperState = event.getState();
		Event.EventType eventType = event.getType();
		if (Event.KeeperState.SyncConnected == keeperState) {
			if (Event.EventType.None == eventType) {
				log.info(LOG_PREFIX_OF_THREAD + "成功连接上ZK服务器");
				connectedSemaphore.countDown();
			} else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
				log.info(LOG_PREFIX_OF_THREAD + "收到情报，排我前面的家伙已挂，我是不是可以出山了？");
				try {
					if (checkMinPath()) {
						getLockSuccess();
					}
				} catch (KeeperException e) {
					e.printStackTrace();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		} else if (Event.KeeperState.Disconnected == keeperState) {
			log.info(LOG_PREFIX_OF_THREAD + "与ZK服务器断开连接");
		} else if (Event.KeeperState.AuthFailed == keeperState) {
			log.info(LOG_PREFIX_OF_THREAD + "权限检查失败");
		} else if (Event.KeeperState.Expired == keeperState) {
			log.info(LOG_PREFIX_OF_THREAD + "会话失效");
		}
	}

	/**
	 * 获取锁
	 * 
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	private void getLock() throws KeeperException, InterruptedException {
		selfPath = zk.create(SUB_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

		log.info(LOG_PREFIX_OF_THREAD + "创建锁路径:{}", selfPath);
		if (checkMinPath()) {
			getLockSuccess();
		}
	}

	/**
	 * 创建节点
	 * 
	 * @param path
	 * @param data
	 * @param needWatch
	 * @return
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	public boolean createPath(String path, String data, boolean needWatch)
			throws KeeperException, InterruptedException {
		if (zk.exists(path, needWatch) == null) {
			String r = zk.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			log.info(LOG_PREFIX_OF_THREAD + "节点创建成功, Path: " + r + ",content: " + data);
		}
		return true;
	}

	/**
	 * 创建ZK连接
	 * 
	 * @param connectString
	 * @param sessionTimeout
	 * @throws IOException
	 * @throws InterruptedException
	 */
	public void createConnection(String connectString, int sessionTimeout) throws IOException, InterruptedException {
		zk = new ZooKeeper(connectString, sessionTimeout, this);
		connectedSemaphore.await();
	}

	/**
	 * 获取锁成功
	 * 
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	private void getLockSuccess() throws KeeperException, InterruptedException {
		if (zk.exists(this.selfPath, false) == null) {
			log.error(LOG_PREFIX_OF_THREAD + "本节点已不存在了。。。");
			return;
		}
		log.info(LOG_PREFIX_OF_THREAD + "获取锁成功，赶紧工作");

		Thread.sleep(2000);
		log.info(LOG_PREFIX_OF_THREAD + "删除本节点" + selfPath);
		zk.delete(this.selfPath, -1);
		releaseConnection();
		threadSemaphore.countDown();
	}

	/**
	 * 关闭zk连接
	 */
	private void releaseConnection() {
		if (this.zk != null) {
			try {
				this.zk.close();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		log.info(LOG_PREFIX_OF_THREAD + "释放连接");
	}

	/**
	 * 检查自己是不是最小的节点
	 * 
	 * @return
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	private boolean checkMinPath() throws KeeperException, InterruptedException {
		List<String> subNodes = zk.getChildren(GROUP_PATH, false);
		Collections.sort(subNodes);
		int index = subNodes.indexOf(selfPath.substring(GROUP_PATH.length() + 1));

		switch (index) {
		case -1: {
			log.error(LOG_PREFIX_OF_THREAD + "本节点点不存在");
			return false;
		}
		case 0: {
			log.info(LOG_PREFIX_OF_THREAD + "子节点中，我是老大");
			return true;
		}
		default: {
			this.waitPath = GROUP_PATH + "/" + subNodes.get(index - 1);
			log.info(LOG_PREFIX_OF_THREAD + "获取子节点中，排在我前面的" + waitPath);

			zk.getData(waitPath, true, new Stat());
		}
		}
		return false;
	}

	public static void main(String[] args) {
		for (int i = 0; i < THREAD_NUM; i++) {
			final int threadId = i + 1;
			new Thread() {
				@Override
				public void run() {
					try {
						DistributedLock dc = new DistributedLock(threadId);
						dc.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
						// GROUP_PATH不存在的话，由一个线程创建即可；
						synchronized (threadSemaphore) {
							dc.createPath(GROUP_PATH, "该节点由线程" + threadId + "创建", true);
						}
						dc.getLock();
					} catch (Exception e) {
						log.error("【第" + threadId + "个线程】 抛出的异常：");
						e.printStackTrace();
					}
				}
			}.start();
		}
		try {
			threadSemaphore.await();
			log.info("所有线程运行结束!");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}
